package concurrency

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlin.time.TimeSource
import net.Net
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.withTimeoutOrNull

private fun isHttpSuccess(code: Int?) = code != null && code in 200..299

private fun pickUrlByModulo(dispatchMs: Long, urls: List<String>): String {
    require(urls.isNotEmpty()) { "requestUrls must not be empty" }
    val idx = ((dispatchMs % urls.size).toInt() + urls.size) % urls.size
    return urls[idx]
}

suspend fun runNCoroutinesMixedVerbose(
    total: Int,
    specialIndex: Int,
    onProgress: suspend (Int) -> Unit,
    onLog: suspend (String) -> Unit,
    onActiveChange: suspend (all: Int, normal: Int, video: Int) -> Unit,
    doNormalWork: suspend () -> Unit,
    doVideoWork: suspend () -> Unit,

    requestUrl: String,
    requestGroupA: Set<Int>,
    requestGroupB: Set<Int>,
    requestGroupC: Set<Int>,
    periodAms: Long,
    periodBms: Long,
    periodCms: Long,
    jitterMs: Long,
    requestRunMs: Long,

    onRequestResult: suspend (
        id: Int,
        group: Char,
        attempt: Int,
        ok: Boolean,
        code: Int?,
        error: String?
    ) -> Unit,

    requestUrls: List<String>? = null
) = coroutineScope {
    val progressMutex = Mutex()
    var counter = 0

    val activeMutex = Mutex()
    var activeAll = 0
    var activeNormal = 0
    var activeVideo = 0

    suspend fun reportActive() {
        val (a, n, v) = activeMutex.withLock { Triple(activeAll, activeNormal, activeVideo) }
        onActiveChange(a, n, v)
    }
    suspend fun inc(isVideo: Boolean) {
        activeMutex.withLock { activeAll++; if (isVideo) activeVideo++ else activeNormal++ }
        reportActive()
    }
    suspend fun dec(isVideo: Boolean) {
        activeMutex.withLock { activeAll--; if (isVideo) activeVideo-- else activeNormal-- }
        reportActive()
    }

    fun pickUrlByModulo(dispatchMs: Long, urls: List<String>): String {
        require(urls.size == 5) { "requestUrls must contain exactly 5 urls (index 0..4)" }
        val idx = ((dispatchMs % urls.size).toInt() + urls.size) % urls.size
        return urls[idx]
    }


    val producerIdsAll: List<Int> = (requestGroupA + requestGroupB + requestGroupC).toList()
    val producers: List<Int> = if (producerIdsAll.size > 100) producerIdsAll.take(100) else producerIdsAll

    val normalAll = (0 until total).filter { it != specialIndex }
    val consumerCandidates = normalAll.filter { it !in producers }
    val consumers: List<Int> = consumerCandidates.take(minOf(100, consumerCandidates.size))


    val resultBoxes: Array<CompletableDeferred<String>?> = Array(total) { null }
    producers.forEach { pid -> resultBoxes[pid] = CompletableDeferred() }

    fun mapConsumerToProducer(consumerId: Int): Int? {
        if (producers.isEmpty()) return null
        val idx = ((consumerId % producers.size) + producers.size) % producers.size
        return producers[idx]
    }

    repeat(total) { id ->
        val isVideo = (id == specialIndex)
        val name = if (isVideo) "video-$id" else "normal-$id"
        val dispatcher = if (isVideo) Dispatchers.IO else Dispatchers.Default

        launch(dispatcher + CoroutineName(name)) {
            val coName = coroutineContext[CoroutineName]?.name ?: name
            onLog("LAUNCH $coName"); inc(isVideo)

            val t = TimeSource.Monotonic.markNow()
            onLog("START  $coName")

            try {
                if (isVideo) {
                    doVideoWork()
                } else {
                    val inA = id in requestGroupA
                    val inB = id in requestGroupB
                    val inC = id in requestGroupC

                    if (!inA && !inB && !inC && id in consumers) {
                        val producerId = mapConsumerToProducer(id)
                        if (producerId == null) {
                            onLog("CONS id=$id no-producer-available -> fallback doNormalWork()")
                            doNormalWork()
                        } else {
                            val box = resultBoxes[producerId]
                            if (box == null) {
                                onLog("CONS id=$id mapped producer=$producerId but no box -> fallback doNormalWork()")
                                doNormalWork()
                            } else {
                                // 等待生产者的首个字符串结果（最多等 requestRunMs）
                                val text = withTimeoutOrNull(requestRunMs) { box.await() }
                                if (text == null) {
                                    onLog("CONS id=$id -> producer=$producerId timeout (${requestRunMs}ms) no result")
                                } else {
                                    val size = text.length
                                    onLog("CONS id=$id <- producer=$producerId got size=$size")
                                }
                                doNormalWork()
                            }
                        }

                    } else if (inA || inB || inC) {
                        val base: Long = when {
                            inA -> periodAms
                            inB -> periodBms
                            else -> periodCms
                        }
                        val group = when {
                            inA -> 'A'
                            inB -> 'B'
                            else -> 'C'
                        }
                        val start = TimeSource.Monotonic.markNow()
                        var loops = 0
                        var producedOnce = false

                        while (isActive && start.elapsedNow().inWholeMilliseconds < requestRunMs) {
                            val delayMs: Long = nextPeriodWithJitter(base, jitterMs)

                            val finalUrl = requestUrls?.let { pickUrlByModulo(delayMs, it) } ?: requestUrl
                            val routeIdx = requestUrls?.let { ((delayMs % it.size).toInt() + it.size) % it.size } ?: 0

                            try {
                                val code = Net.getStatus(finalUrl)
                                val ok = isHttpSuccess(code)

                                val text = "id=$id group=$group loop=${loops + 1} url[$routeIdx]=$finalUrl HTTP $code"

                                onLog("REQ  id=$id group=$group delay=${delayMs}ms routeIdx=$routeIdx url=$finalUrl -> HTTP $code")
                                onRequestResult(id, group, loops + 1, ok, code, null)

                                if (!producedOnce && id in producers) {
                                    resultBoxes[id]?.let { box ->
                                        if (!box.isCompleted) {
                                            box.complete(text)
                                            producedOnce = true
                                            onLog("PROD id=$id -> box completed (len=${text.length})")
                                        }
                                    }
                                }
                            } catch (e: Throwable) {
                                onLog("REQ  id=$id group=$group delay=${delayMs}ms routeIdx=$routeIdx url=$finalUrl " +
                                        "ERROR ${e::class.simpleName}: ${e.message}")
                                onRequestResult(id, group, loops + 1, false, null, e.message)

                                if (!producedOnce && id in producers) {
                                    val errText = "id=$id group=$group ERROR ${e::class.simpleName}: ${e.message ?: "unknown"}"
                                    resultBoxes[id]?.let { box ->
                                        if (!box.isCompleted) {
                                            box.complete(errText)
                                            producedOnce = true
                                            onLog("PROD id=$id -> box completed with error text (len=${errText.length})")
                                        }
                                    }
                                }
                            }

                            loops++
                            delay(delayMs)
                        }

                        onLog("REQ  id=$id finished loops=$loops run=${start.elapsedNow().inWholeMilliseconds}ms")
                        if (!producedOnce && id in producers) {
                            val fallback = "NO_RESULT id=$id group=$group"
                            resultBoxes[id]?.let { box ->
                                if (!box.isCompleted) {
                                    box.complete(fallback)
                                    onLog("PROD id=$id -> box fallback text (len=${fallback.length})")
                                }
                            }
                        }

                    } else {
                        doNormalWork()
                    }
                }

                val elapsed = t.elapsedNow().inWholeMilliseconds
                val v = progressMutex.withLock { ++counter }
                onLog("END    $coName elapsed=${elapsed}ms progress=$v/$total")
                onProgress(v)

            } catch (e: CancellationException) {
                onLog("CANCEL $coName"); throw e
            } catch (e: Throwable) {
                val v = progressMutex.withLock { ++counter }
                onLog("ERROR  $coName -> ${e::class.simpleName}: ${e.message} progress=$v/$total")
                onProgress(v)
            } finally {
                dec(isVideo)
            }
        }
    }
}
private fun nextPeriodWithJitter(baseMs: Long, jitterMs: Long): Long {
    val delta = kotlin.random.Random.nextLong(-jitterMs, jitterMs + 1L)
    return (baseMs + delta).coerceAtLeast(0L)
}
